package day03

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 2.4 Transformation转换算子
 *
 * RDD整体上分为Value类型、双Value类型和Key-Value类型
 */
object Transformation01Value {

}

/**
 * 2.4.1 value类型
 *
 * 2.4.1.1 map()
 */
object Transformation_Value_map {

  def main(args: Array[String]): Unit = {
    // 1、创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    // 2、创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 3、创建一个RDD
    val rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)

    // 4、调用 map()方法，每个元素乘以2
    val mapRdd: RDD[Int] = rdd.map(_ * 2)

    // 5、打印输出
    mapRdd.collect().foreach(println)

    // 6、关闭连接
    sc.stop()
  }

}

/**
 * 2.4.1.2	mapPartitions()以分区为单位执行Map
 */
object Transformation_Value_mapPartitions {

  def main(args: Array[String]): Unit = {

    // 1、创建SparkConf并设置App名称
    val sparkConf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    // 2、创建SparkContext，该对象是提交Spark App 的入口
    val sc: SparkContext = new SparkContext(sparkConf)

    // 3、创建一个RDD
    val rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)

    // 4、调用mapPartitions()方法
    val newRdd: RDD[Int] = rdd.mapPartitions(iterator => iterator.map(elem => elem * 2))

    // 5、打印输出
    newRdd.collect().foreach(println)

    // 6、关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.3	mapPartitionsWithIndex()带分区号
 */
object Transformation_Value_mapPartitionsWithIndex {

  def main(args: Array[String]): Unit = {

    // 1、创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    // 2、创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 3、创建RDD
    val rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)

    // 4、调用mapPartitionsWithIndex()
    val newRdd: RDD[(Int, Int)] = rdd.mapPartitionsWithIndex((index, iterator) => {
      iterator.map(elem => (index, elem))
    })

    // 简化版
    val indexRdd: RDD[(Int, Int)] = rdd.mapPartitionsWithIndex((index, items) => {
      items.map((index, _))
    })

    // 5、打印输出
    newRdd.collect().foreach(println)
    println("------------------------------")
    indexRdd.collect().foreach(println)

    // 6、关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.5	flatMap()压平
 */
object Transformation_Value_flatMap {

  def main(args: Array[String]): Unit = {

    //1、创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2、创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3、创建rdd
    val rdd: RDD[List[Int]] = sc.makeRDD(List(List(1, 2), List(3, 4), List(5, 6), List(7)), 2)

    //4、调用flatMap()
    val newRdd: RDD[Int] = rdd.flatMap(list => list)

    //5、打印输出
    newRdd.collect().foreach(println)

    //6、关闭连接
    sc.stop()
  }

}

/**
 * 2.4.1.6	glom()分区转换数组
 */
object Transformation_Value_glom {

  def main(args: Array[String]): Unit = {

    //1、创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2、创建SparkContext,该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3、创建一个rdd
    val rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)

    //4、调用glom()，将每个分区的数据变成一个数组
    val newRdd: RDD[Array[Int]] = rdd.glom()

    //5、调用map()，获取每个分区的最大值。0->1,2   1->3,4
    val newRdd1: RDD[Int] = newRdd.map(array => array.min)

    //6、打印输出
    newRdd1.collect().foreach(println)

    //7、关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.7	groupBy()分组
 *
 * groupBy会存在shuffle过程
 * shuffle：将不同的分区数据进行打乱重组的过程
 * shuffle一定会落盘。可以在local模式下执行程序，通过4040看效果。(127.0.0.1:4040)
 *
 */
object Transformation_Value_groupBy {
  def main(args: Array[String]): Unit = {

    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3、创建rdd
    val rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)

    //4、调用groupBy()
    val newRdd: RDD[(Int, Iterable[Int])] = rdd.groupBy(elem => elem % 2)

    //5、打印输出
    newRdd.collect().foreach(println)

    val rdd1: RDD[String] = sc.makeRDD(List("hello", "hive", "hadoop", "spark", "scala"))
    rdd1.groupBy(str => str.substring(0, 1)).collect().foreach(println)

    Thread.sleep(100000)

    //6、关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.8	GroupBy之WordCount
 */
object Transformation_Value_groupBy01 {

  def main(args: Array[String]): Unit = {

    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3、创建RDD
    val strList: List[String] = List("Hello Scala", "Hello Spark", "Hello World")
    val rdd: RDD[String] = sc.makeRDD(strList)

    //4、将字符串拆分成一个一个的单词
    val wordRdd: RDD[String] = rdd.flatMap(str => str.split(" "))

    //5、将单词结果进行转换
    val wordToOneRdd: RDD[(String, Int)] = wordRdd.map(word => (word, 1))

    //6、将转换结构后的数据分组
    val groupRdd: RDD[(String, Iterable[(String, Int)])] = wordToOneRdd.groupBy(t => t._1)

    //7、将分组后的数据进行结构的转换
    val wordToSum: RDD[(String, Int)] = groupRdd.map {
      case (word, list) => (word, list.size)
    }

    wordToSum.collect().foreach(println)


    //4.关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.8	GroupBy之WordCount复杂版
 */
object Transformation_Value_groupBy02 {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("Hello Scala", 2), ("Hello Spark", 3), ("Hello World", 2)))

    val newRdd: RDD[(String, Int)] = rdd.flatMap {
      case (str, num) =>
        str.split(" ").map(str => (str, num))
    }

    val newRdd1: RDD[(String, Iterable[(String, Int)])] = newRdd.groupBy(tuple => tuple._1)

    val newRdd2: RDD[(String, Int)] = newRdd1.map {
      case (str, tuples) =>
        var i: Int = 0;
        for (elem <- tuples) {
          i = i + elem._2
        }
        (str, i)
    }

    newRdd2.collect().foreach(println)

    //4.关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.9	filter()过滤
 */
object Transformation_Value_filter {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3、创建RDD
    val rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)

    //4、过滤符合条件的数据
    val newRdd: RDD[Int] = rdd.filter(elem => elem % 2 == 0)

    //5、打印输出
    newRdd.collect().foreach(println)

    //4.关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.10	sample()采样
 *
 * def sample {
 *
 * withReplacement: Boolean, // withReplacement表述：抽出的数据是否放回，true为有放回的抽样，false为无放回的抽样
 * fraction: Double, // 当withReplacement=false时：选择每个元素的概率；取值一定是[0,1];
 * // 当withReplacement=true是：选择每个元素的期望次数；取值必须大于等于0
 * seed: Long = Utils.random.nextLong):RDD[T] // seed表示：指定随机数生成器种子
 * }
 */
object Tansformation_Value_sample {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val rdd: RDD[Int] = sc.makeRDD(1 to 10)

    //抽样放回，fraction表示每个元素的期望次数
    rdd.sample(withReplacement = true, fraction = 2, seed = 2).collect().foreach(println)

    println("-------------------------------------------------")

    //抽样不放回，fraction表示每个元素的出现概率。取值必须为[0,1]
    rdd.sample(withReplacement = false, fraction = 0.5, seed = 2).collect().foreach(println)

    //4.关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.11	distinct()去重。效率比HaseSet集合的方式要高
 */
object Transformation_Value_distinct {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3、创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 1, 5, 2, 4, 5, 3), 3)

    // 打印去重前的数据
    rdd.mapPartitionsWithIndex((index, iterable) => {
      println(index + "号分区：" + iterable.mkString(", "))
      iterable
    }).collect()

    println("----------------------------------------")

    // 调用distinct()
    val newRdd: RDD[Int] = rdd.distinct()

    // 打印去重后的数据
    newRdd.mapPartitionsWithIndex((index, iterable) => {
      println(index + "号分区：" + iterable.mkString(", "))
      iterable
    }).collect()

    println("----------------------------------------")

    // 指定去重后分区的个数
    val newRdd1: RDD[Int] = rdd.distinct(2)

    // 打印去重后的数据
    newRdd1.mapPartitionsWithIndex((index, iterable) => {
      println(index + "号分区：" + iterable.mkString(", "))
      iterable
    }).collect()

    //4.关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.12	coalesce()重新分区
 *
 * Coalesce算子包括：配置执行Shuffle和配置不执行Shuffle两种方式。
 *
 * 注意：默认情况下，不执行shuffle。如果使用coalesce扩大分区是不起作用的
 *
 * 不带shuffle容易产生数据倾斜
 */
object Transformation_Value_coalesce_without_shuffle {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3、创建RDD
    val rdd: RDD[Int] = sc.makeRDD(1 to 4, 4)

    // 打印缩减分区前的数据
    rdd.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区：" + datas.mkString(", "))
      datas
    }).collect()

    println("------------------------------------------------")

    //4、缩减分区
    val coalesceRdd: RDD[Int] = rdd.coalesce(2)

    //5、打印输出每个分区的数据
    coalesceRdd.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区：" + datas.mkString(", "))
      datas
    }).collect()


    //4.关闭连接
    sc.stop()
  }
}

/**
 * 执行Shuffle方式
 *
 * 问题：假如求shuffle后，分区0中的所有数据的和，需要等待读取所有分区的数据后，才能求和。
 * 如果数据一直放在内存中，容易导致OOM
 *
 * 解决办法：shuffle，必须要落盘等待数据的到来
 */
object Transformation_Value_coalesce_with_shuffle {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3、创建RDD
    val rdd: RDD[Int] = sc.makeRDD(1 to 6, 3)

    //打印缩减分区前的数据
    rdd.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区：" + datas.mkString(", "))
      datas
    }).collect()

    //4、缩减分区，带shuffle
    val newRdd: RDD[Int] = rdd.coalesce(2, shuffle = true)

    // 打印缩减分区后的数据
    newRdd.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区：" + datas.mkString(", "))
      datas
    }).collect()

    //4.关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.13	repartition()重新分区（执行Shuffle）
 *
 * 该操作内部执行的是coalesce操作，coalesce(numPartitions, shuffle = true)
 *
 * 可以将分区变少，也可以将分区变多
 */
object Transformation_Value_repartition {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3、创建RDD
    val rdd: RDD[Int] = sc.makeRDD(1 to 10, 4)

    //打印重新分区前的数据
    rdd.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区：" + datas.mkString(", "))
      datas
    }).collect()

    println("-----------------------------------------")

    //4、重新分区
    // rdd.coalesce(2, shuffle = true)
    val newRdd: RDD[Int] = rdd.repartition(2)

    // 打印重新分区后的数据
    newRdd.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区： " + datas.mkString(", "))
      datas
    }).collect()

    //4.关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.14	coalesce和repartition区别
 *
 * ① coalesce重新分区，可以选择是否进行shuffle过程。由参数shuffle: Boolean = false/true决定。
 * ② repartition实际上是调用的coalesce，进行shuffle。源码如下：
 * def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
 * coalesce(numPartitions, shuffle = true)
 * }
 * ③ coalesce一般用于缩减分区，如果扩大分区，不使用shuffle是没有意义的，repartition扩大分区执行shuffle。
 *
 */
object Transformation_Value_coalesce_repartition {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 3、 创建一个RDD
    val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11), 4)

    // 打印合并分区之前的数据
    rdd.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区：" + datas.mkString(", "))
      datas
    }).collect()

    println("---------------------------------------------")

    // 4、合并分区(没有shuffle)
    val newRdd: RDD[Int] = rdd.coalesce(2)

    // 打印合并分区之后的数据
    newRdd.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区：" + datas.mkString(", "))
      datas
    }).collect()

    println("---------------------------------------------")

    //5、重新分区(有shuffle)
    val newRdd1: RDD[Int] = rdd.repartition(2)
    newRdd1.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区：" + datas.mkString(", "))
      datas
    }).collect()

    //4.关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.15	sortBy()排序
 */
object Transformation_Value_sortBy {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3、创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(2, 1, 3, 4, 6, 5))

    //4、调用sortBy()。默认为升序
    //    val newRdd: RDD[Int] = rdd.sortBy(num => num, ascending = true)
    val newRdd: RDD[Int] = rdd.sortBy(num => num)
    newRdd.collect().foreach(println)

    println("-------------------------------------------------------")

    //5、倒序
    val newRdd1: RDD[Int] = rdd.sortBy(num => num, ascending = false)
    newRdd1.collect().foreach(println)

    println("-------------------------------------------------------")

    //6、创建一个RDD
    val strRdd: RDD[String] = sc.makeRDD(List("1", "22", "12", "2", "3"))

    //7、按照字符的int值排序
    val strSortedRdd: RDD[String] = strRdd.sortBy(_.toInt)

    strSortedRdd.collect().foreach(println)

    println("-------------------------------------------------------")

    //8、 创建一个RDD
    val rdd3: RDD[(Int, Int)] = sc.makeRDD(List((2, 1), (1, 2), (1, 1), (2, 2)))

    //9、先按照tuple的第一个值排序，相等再按照第2个值排
    rdd3.sortBy(t => t).collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.4.1.16	pipe()调用脚本
 *
 * 管道，针对每个分区，都调用一次shell脚本，返回输出的RDD
 *
 * 注意：在Worker节点可以访问到的位置放脚本
 */
object Transformation_Value_pipe {
  def main(args: Array[String]): Unit = {

  }
}





